package com.meihua.code.thread;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.qingqing.common.async.qqmq.QqmqMsgConvert;
import com.qingqing.common.async.rabbit.AsyncEventNotifyRabbitMsgEntity.AsyncEventNotifyRabbitMsgBody;
import com.qingqing.common.rabbit.bean.RabbitMessageBody;
import com.qingqing.common.rabbit.bean.RabbitMessageEntity;
import com.qingqing.common.rabbit.bean.RabbitMessageHeader;
import com.qingqing.common.rabbit.bean.RabbitMsgEntityRouteWarpper;
import com.qingqing.common.util.JsonUtil;
import com.qingqing.common.web.aspect.log.InheritedGuid;
import com.qingqing.common.web.aspect.log.InheritedGuid.Ops;
import com.qingqing.common.web.util.GuidUtil;
import com.qingqing.qqmq.base.producer.builder.MessageBuilder;
import com.qingqing.qqmq.service.QQMQProduceService;
import com.qingqing.qqmq.thrift.Result;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;


/**
 * 异步qqMq弹性发送者
 *
 * @author humeihua
 * @date 2021/07/21
 */
public  class AsyncQqMqElasticSender implements com.qingqing.api.rabbit.AsyncMqRouteSender {

    private static final Logger logger = LoggerFactory.getLogger(AsyncQqMqElasticSender.class);

    private ThreadPoolExecutor poolExecutor;

    private String senderName;

    private int maximumPoolSize;

    private int capacity;

    private int corePoolSize=1;
    /**
     * 启用同步发送
     **/
    private boolean enableSyncSending = false;

    private final QQMQProduceService qqmqProduceService;

    private final String topicName;

    private QqmqMsgConvert convert;

    private Map<Integer,Integer> elasticScoreMap;

    public AsyncQqMqElasticSender(QQMQProduceService qqmqProduceService,
                                  String topicName,
                                  QqmqMsgConvert qqmqMsgConvert,
                                  String  senderName){
        this.senderName=senderName;
        this.qqmqProduceService = qqmqProduceService;
        this.topicName = topicName;
        this.convert = qqmqMsgConvert;
        this.maximumPoolSize = 5;
        this.capacity = 10000;
    }

    public AsyncQqMqElasticSender(QQMQProduceService qqmqProduceService,
                                  String topicName,
                                  QqmqMsgConvert qqmqMsgConvert,
                                  String  senderName,
                                  int capacity,
                                  int corePoolSize,
                                  int maximumPoolSize,
                                  boolean enableSyncSending){
        this.senderName=senderName;
        this.qqmqProduceService = qqmqProduceService;
        this.topicName = topicName;
        this.corePoolSize=corePoolSize;
        this.convert = qqmqMsgConvert;
        this.maximumPoolSize = maximumPoolSize>0?maximumPoolSize:1;
        this.capacity = capacity;
        this.enableSyncSending=enableSyncSending;
    }

    public void elastic(){
        //当前弹性分数
        Integer elasticScore = elasticScoreMap.get(corePoolSize);
        int size = poolExecutor.getQueue().size();
        if (elasticScore<size&&corePoolSize<maximumPoolSize){
            logger.info("elasticScore:{} ,size: {}  add Thread",elasticScore,size ,poolExecutor.getPoolSize());
            corePoolSize++;
            poolExecutor.setCorePoolSize(corePoolSize);
        }else if (elasticScore>size&&corePoolSize>0){
            corePoolSize--;
            poolExecutor.setCorePoolSize(corePoolSize);
        }
    }

    public void daemon(){
        Thread thread = new Thread(()->{
            while (true){
                try {
                    elastic();
                    Thread.sleep(5000);
                    logger.info(" PoolSize:{}, CorePoolSize: {} ,  MaximumPoolSize : {}  taskNum:{} ",poolExecutor.getPoolSize(),poolExecutor.getCorePoolSize(),poolExecutor.getMaximumPoolSize(),poolExecutor.getTaskCount());
                }catch (Exception e){
                    logger.error("daemon error",e);
                }
            }
        });
        thread.setDaemon(true);
        thread.start();
    }

    interface ElasticExecutionHandler {

        void handler();

    }


    class AbortPolicy implements ElasticExecutionHandler {

        public AbortPolicy() { }

        @Override
        public void handler() {
            Integer elasticScore = elasticScoreMap.get(corePoolSize);
            if (elasticScore<poolExecutor.getQueue().size()){
                corePoolSize++;
                poolExecutor.setCorePoolSize(corePoolSize);
            }else if (corePoolSize>0){
                corePoolSize--;
                poolExecutor.setCorePoolSize(corePoolSize);
            }
        }
    }

    public void start() {
        elasticScoreMap = new HashMap<>(maximumPoolSize);
        for (int i=0;i<=maximumPoolSize;i++){
            elasticScoreMap.put(i,(capacity/maximumPoolSize)*i);
        }
        logger.info("{} start ,maximumPoolSize:{}, elasticScoreMap: {}",senderName,maximumPoolSize,elasticScoreMap);
        stop();
        poolExecutor =  new ThreadPoolExecutor(corePoolSize, maximumPoolSize, 1, TimeUnit.SECONDS, new ArrayBlockingQueue<>(capacity),
                new ThreadFactoryBuilder().setNameFormat(senderName+"-pool-%d").build(),
                (runnable,executor)->{
                    try {
                        elastic();
                        SendTask task = (SendTask) runnable;
                        if (enableSyncSending &&task.msg.getEntity()!=null && ! (task.msg.getEntity() instanceof AsyncEventNotifyRabbitMsgBody)){
                            convertAndSenderMessage(task.msg);
                           // logger.warn("mq msg queue is full. sender mq message sync :{}", JsonUtil.getJsonFromObject(task.msg));
                        }else {
                           // logger.error("mq msg queue is full. sender mq message discard policy :{}", JsonUtil.getJsonFromObject(task.msg));
                        }
                    }catch (Exception e){
                       logger.error("mq rejected execution handler error ",e);
                    }
                });
        poolExecutor.allowCoreThreadTimeOut(true);
        daemon();
        logger.info("qqmq async message thread has started, senderName:{}-{}", senderName);
    }


    class SendTask implements Runnable {

        private RabbitMsgEntityRouteWarpper msg;

        public SendTask(RabbitMsgEntityRouteWarpper wrapper) {this.msg = wrapper;}

        @Override
        public void run() {
            GuidUtil.tryUseGuid(msg);
            convertAndSenderMessage(msg);
        }
    }


    public void convertAndSenderMessage(RabbitMsgEntityRouteWarpper wrapper){
        if(true){
            //   logger.info(Thread.currentThread().getName()+"模拟发送");
            try {
                Thread.sleep(10);
            }catch (Exception e){

            }
            return;
        }
        try {
            if( logger.isDebugEnabled() ) {
                logger.debug("send qqmq msg: {}", JsonUtil.getJsonFromObject(wrapper));
            }
            MessageBuilder messageBuilder = qqmqProduceService.messageBuilder()
                    .setTopic(topicName)
                    .setBody(convertMessage(wrapper))
                    .setTags(convertTags(wrapper.getTags()));

            if(!StringUtils.isBlank(wrapper.getRouteKey())){
                messageBuilder.setRoutingKey(wrapper.getRouteKey());
            }

            Result r = messageBuilder.send();
            if( r.code != 0 ) {
                logger.error("send qqmq msg error, ignore it. send resultCode:{}, resultKey:{}, resultMsg:{}. msg:{}",
                        r.getCode(), r.getKey(), r.getMsg(), JsonUtil.getJsonFromObject(wrapper));
            }
        } catch (RuntimeException e) {
            logger.error("send qqmq msg error, ignore it. msg:{}", JsonUtil.getJsonFromObject(wrapper), e);
        }
    }

    private String convertMessage(RabbitMsgEntityRouteWarpper wrapper){
        RabbitMessageBody body = wrapper.getEntity();
        RabbitMessageHeader header = convert.formMsgHeader(body.getVersion());
        RabbitMessageEntity entity = new RabbitMessageEntity(header, body);
        return JsonUtil.getJsonFromObject(entity);
    }

    public void stop() {
        if(poolExecutor != null){
            logger.info("try to stop AsyncQqmqSender.");
            poolExecutor.shutdownNow();
            logger.info("stop AsyncQqmqSender suc.");
        }
    }


    @Override
    public void asyncSendMsg(RabbitMsgEntityRouteWarpper msg) {
        poolExecutor.execute(new SendTask(msg));
    }

    @Override
    @InheritedGuid(ops = Ops.save)
    public void syncSendMsg(RabbitMsgEntityRouteWarpper msg) {
        try {
            logger.info("sync send");
            convertAndSenderMessage(msg);
        } catch (RuntimeException ignore) {
            logger.error("send qqmq msg to queue error. msg:{}", JsonUtil.getJsonFromObject(msg), ignore);
        }
    }


    private static String [] convertTags(String... values){
        if(values!=null){
            if (values.length<=0){
                return null;
            }
            for(String value:values){
                if(StringUtils.isNotBlank(value)){
                    return values;
                }
            }
            return null;
        }
        return null;
    }

    public void setConvert(QqmqMsgConvert convert) {
        this.convert = convert;
    }

    public static void main(String[] args) {
        AsyncQqMqElasticSender asyncQqMqElasticSender = new AsyncQqMqElasticSender(null,
                "tst",null,
                "测试qqmq",
                10000,
                1,
                10,
                true);
        asyncQqMqElasticSender.start();

        long l = System.currentTimeMillis();

        for (int i=0;i<20000;i++){
            try {
                Thread.sleep(3);
                RabbitMsgEntityRouteWarpper warpper= new RabbitMsgEntityRouteWarpper();
                AsyncEventNotifyRabbitMsgBody body = new AsyncEventNotifyRabbitMsgBody();
                warpper.setMsgBody(body);
                body.setRef_id("1");
                asyncQqMqElasticSender.asyncSendMsg(warpper);
            }catch (Exception e){

            }

        }
        asyncQqMqElasticSender.stop();
        logger.info("done 耗时："+(System.currentTimeMillis()-l));
    }

}

